package com.gugee.tiktok.data.consumer.fansportrait.listener;

import com.gugee.tiktok.data.common.msg.FansPortraitMessage;
import com.gugee.tiktok.data.consumer.fansportrait.config.FansPortraitAppConfig;
import com.gugee.tiktok.data.consumer.fansportrait.core.main.SaveFansPortraitMain;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

/**
 * Description:
 * User: Huang xin
 * Date: 2020-11-12 18:06
 */
@Component
@Slf4j
public class FansPortraitMessageListener {
    @Autowired
    SaveFansPortraitMain saveFansPortraitMain;

    @Autowired
    FansPortraitAppConfig fansPortraitAppConfig;
    @KafkaListener(id="${kafka.consumer.group.fansportrait.id}",topics = {"${kafka.topics.fansportrait.online}"},
            containerFactory = "factoryFansPortraitAck")
    public void listen(List<FansPortraitMessage> messages, Acknowledgment ack) {
        long start = System.currentTimeMillis();
        long acceptMsgTimeSecRange = fansPortraitAppConfig.getAcceptMsgTimeMinRange() * 60 * 1000;
        String error = "success";
        try {

            long isRange=start-acceptMsgTimeSecRange;
            final int[] noInRangeCount = {0};
            /**
             * 消息如果不在处理时限内，则丢弃
             */
            List<FansPortraitMessage> fpms = messages.stream().filter(msg->{
                if(msg.getEventTime() < isRange){
                    noInRangeCount[0]++;
                    return false;
                }
                if(msg.getId() == null || msg.getId() == 0){
                    log.error("接收到的消息无id值，awemeId={}",msg.getAwemeId());
                    return false;
                }
                return true;
            }).collect(Collectors.toList());
            if(noInRangeCount[0] > 0){
                log.info("接收到fansPortrait消息数量: {},需解析fansPortrait消息数量： {},不在处理时限内消息数量： {},时限分钟：{}" , messages.size(),fpms.size(),
                        noInRangeCount[0],fansPortraitAppConfig.getAcceptMsgTimeMinRange());
            }

            saveFansPortraitMain.handle(fpms);

            ack.acknowledge();
        }catch (Exception e){
            log.error("发生异常：{}" ,e.getMessage());
            error=e.getMessage();
        }finally {

            long times=(System.currentTimeMillis()-start)/1000;
            if(times<1){
                times=1;
            }
            log.info("fansPortrait消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,结果：{}" , messages.size(),times,messages.size()/times,error);
        }
    }

}
